package cn.metona.mq.producer;

import cn.metona.mq.config.ConfigManager;
import cn.metona.mq.core.Message;
import cn.metona.mq.core.MessageQueue;
import cn.metona.mq.exception.MessageRejectedException;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;

/**
 * 消息生产者
 * 支持异步发送、批量发送、流控等功能
 */
public class MessageProducer {
    private static final Logger logger = Logger.getLogger(MessageProducer.class.getName());

    private final String producerId;                    // 生产者ID
    private final MessageQueue queue;                   // 关联的队列
    private final ExecutorService executorService;      // 异步执行线程池
    private final AtomicInteger pendingMessages;        // 待处理消息计数
    private final int maxPendingMessages;               // 最大待处理消息数

    /**
     * 构造函数
     */
    public MessageProducer(String producerId, MessageQueue queue) {
        if (producerId == null || producerId.isEmpty()) {
            throw new IllegalArgumentException("Producer ID cannot be null or empty");
        }
        if (queue == null) {
            throw new IllegalArgumentException("Queue cannot be null");
        }

        this.producerId = producerId;
        this.queue = queue;
        // 配置管理器
        ConfigManager configManager = ConfigManager.getInstance();
        this.pendingMessages = new AtomicInteger(0);
        this.maxPendingMessages = configManager.getIntProperty("mq.producer.max.pending", 1000);

        // 创建自定义线程工厂
        ThreadFactory threadFactory = new ThreadFactory() {
            private final AtomicInteger threadNumber = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "Producer-" + producerId + "-" + threadNumber.getAndIncrement());
                thread.setDaemon(false);
                return thread;
            }
        };

        this.executorService = Executors.newCachedThreadPool(threadFactory);
    }

    /**
     * 同步发送消息
     */
    public String sendMessage(Message message) {
        if (message == null) {
            throw new IllegalArgumentException("Message cannot be null");
        }

        // 流控检查
        flowControlCheck();

        queue.sendMessage(message);
        return message.messageId();
    }

    /**
     * 异步发送消息
     */
    public CompletableFuture<String> sendMessageAsync(Message message) {
        if (message == null) {
            throw new IllegalArgumentException("Message cannot be null");
        }

        // 流控检查
        flowControlCheck();

        CompletableFuture<String> future = new CompletableFuture<>();
        pendingMessages.incrementAndGet();

        executorService.submit(() -> {
            try {
                queue.sendMessage(message);
                future.complete(message.messageId());
            } catch (Exception e) {
                future.completeExceptionally(e);
            } finally {
                pendingMessages.decrementAndGet();
            }
        });

        return future;
    }

    /**
     * 批量发送消息（同步）
     */
    public void sendMessages(List<Message> messages) {
        if (messages == null) {
            throw new IllegalArgumentException("Messages list cannot be null");
        }

        // 流控检查
        flowControlCheck(messages.size());

        queue.sendMessages(messages);
    }

    /**
     * 批量发送消息（异步）
     */
    public CompletableFuture<Void> sendMessagesAsync(List<Message> messages) {
        if (messages == null) {
            throw new IllegalArgumentException("Messages list cannot be null");
        }

        // 流控检查
        flowControlCheck(messages.size());

        CompletableFuture<Void> future = new CompletableFuture<>();
        pendingMessages.addAndGet(messages.size());

        executorService.submit(() -> {
            try {
                queue.sendMessages(messages);
                future.complete(null);
            } catch (Exception e) {
                future.completeExceptionally(e);
            } finally {
                pendingMessages.addAndGet(-messages.size());
            }
        });

        return future;
    }

    /**
     * 流控检查
     */
    private void flowControlCheck() {
        flowControlCheck(1);
    }

    /**
     * 流控检查（批量）
     */
    private void flowControlCheck(int messageCount) {
        if (messageCount <= 0) {
            return;
        }

        int currentPending = pendingMessages.get();
        if (currentPending + messageCount > maxPendingMessages) {
            throw new MessageRejectedException("Producer flow control exceeded: " +
                    (currentPending + messageCount) + " > " + maxPendingMessages);
        }
    }

    /**
     * 获取生产者ID
     */
    public String getProducerId() {
        return producerId;
    }

    /**
     * 获取关联队列
     */
    public MessageQueue getQueue() {
        return queue;
    }

    /**
     * 获取待处理消息数
     */
    public int getPendingMessageCount() {
        return pendingMessages.get();
    }

    /**
     * 关闭生产者
     */
    public void close() {
        executorService.shutdown();
        logger.info("Producer closed: " + producerId);
    }
}